package cn.getech.data.development.controller.real;

import cn.getech.data.development.constant.PermissionTypeEnum;
import cn.getech.data.development.constant.RealTableType;
import cn.getech.data.development.entity.ConfConnect;
import cn.getech.data.development.entity.TableRegisterEntity;
import cn.getech.data.development.entity.flink.BdpRealTableInfo;
import cn.getech.data.development.entity.permission.RealTableFieldInfoPermission;
import cn.getech.data.development.entity.permission.RealTablePermission;
import cn.getech.data.development.entity.permission.TenantRealTablePermission;
import cn.getech.data.development.model.req.search.DataSearchReq;
import cn.getech.data.development.model.vo.BdpRealTableInfoVO;
import cn.getech.data.development.service.BdpRealTableInfoService;
import cn.getech.data.development.service.ConfConnectService;
import cn.getech.data.development.service.DataPermissionService;
import cn.getech.data.development.service.TableRegisterService;
import cn.getech.data.development.utils.DataPermissionUtil;
import cn.getech.data.development.utils.kafka.LocalKafkaUtil;
import cn.getech.data.intelligence.common.utils.PageUtils;
import cn.getech.data.intelligence.common.utils.R;
import cn.getech.system.center.annotation.SysLog;
import cn.getech.system.center.constant.OperationModulesEnum;
import cn.getech.system.center.constant.OperationTypeEnum;
import cn.getech.system.center.constant.module.ModuleEnum;
import cn.getech.system.center.utils.ShiroUtils;
import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiParam;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.web.bind.annotation.*;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;

@Slf4j
@Api(tags = "实时任务操作日志")
@RestController
@RequestMapping("/bdp/realTime/table")
public class RealTimeTableController {

    @Autowired
    private BdpRealTableInfoService realTableInfoService;
    @Autowired
    private ConfConnectService confConnectService;
    @Autowired
    private DataPermissionService dataPermissionService;
    @Autowired
    private TableRegisterService tableRegisterService;
    @ApiOperation(value = "保存实时表")
    @SysLog(value="保存实时表",type= OperationTypeEnum.UPDATECODE,
            modulesType = OperationModulesEnum.SYSMODECODE,platformModuleType = ModuleEnum.DATA_MAP_MODULE_CODE)
    @PostMapping("/save")   //  /data-development/bdp/realTime/table/save
    public R save(@RequestBody BdpRealTableInfoVO param) {
        String msg = param.verificationParam();
        if (StringUtils.isNotEmpty(msg)) {
            return R.error(msg);
        }
        if (param.getId() != null) { //编辑校验权限
            String res=realTableInfoService.checkPermission(param);
            if (StringUtils.isNotEmpty(res)) {
                return R.error(res);
            }
        }
        int n=realTableInfoService.save(param);
        if (n == -1) {//表名已存在
            return R.error(param.getDbName()+"库下的表名["+param.getTableName()+"]已存在");
        }
        if (n == -2) {//topic名已存在
            return R.error(param.getDbName()+"库下的Topic名 ["+param.getTopicName()+"] 已存在");
        }
        if (n == 0) {
            new Thread(() -> {
                if (RealTableType.INNER.getCode().equals(param.getTableType())) {
                    LocalKafkaUtil.createTopic(param.getTopicName(),
                            param.getPartitionNum(), param.getDuplicateNum(), kafkaUrl);
                }
            }).start();
        }
        return R.okWithData(param);
    }

    /**
     * 获取数据表信息列表
     */
    @ApiOperation("根据角色获取实时表信息列表")
    @GetMapping("/listByRole")
    public R listByRole(@ApiParam(value = "{'table_name':'test','db_name':'test'}") @RequestParam Map<String, Object> params) {
        PageUtils <RealTablePermission> page = realTableInfoService.listPageByRole(params);
        return R.okWithPage(page);
    }

    /**
     * 获取数据表信息列表
     */
    @ApiOperation("根据租户获取实时表信息列表")
    @GetMapping("/listByTenant")
    //@SysLog(value = "根据租户获取实时表信息列表", type = OperationTypeEnum.SELECTCODE, modulesType = OperationModulesEnum.TABLEMODELCODE)
    public R listByTenant(@ApiParam(value = "{'table_name':'test','db_name':'test'}") @RequestParam Map<String, Object> params) {
        if (params.get("tenant_id") == null) {
            return R.error("请选择租户!");
        }
        PageUtils <TenantRealTablePermission> page = realTableInfoService.listPageByTenant(params);
        return R.okWithPage(page);
    }
    /**
     * 保存数据仓库表的权限信息
     */
    @ApiOperation("保存数据仓库表的权限信息")
    @PostMapping("/savePermission")
    @SysLog(value="实时数仓授权",type= OperationTypeEnum.APPROVECODE,
            modulesType = OperationModulesEnum.SYSMODECODE,platformModuleType = ModuleEnum.AUTH_MANAGER_MODULE_CODE)
    public R savePermission( @RequestBody  List<RealTablePermission> dataList) {
//        if (CollectionUtil.isNotEmpty(dataList)) {
//            List<Integer> dataIds = DataPermissionUtil.getDataIds(ShiroUtils.getUserId().intValue(),
//                    PermissionTypeEnum.REAL_TABLE.getType(),
//                    PermissionTypeEnum.ASSIGN.getType());
//            List<Integer> fieldIds = DataPermissionUtil.getDataIds(ShiroUtils.getUserId().intValue(),
//                    PermissionTypeEnum.REAL_TABLE_FILED.getType(),
//                    PermissionTypeEnum.ASSIGN.getType());
//            List<Integer> saveTableIds = dataList.stream().map(i -> i.getDataId()).collect(Collectors.toList());
//            List<Integer> saveFieldIds=new ArrayList<>();
//            for(RealTablePermission realTablePermission:dataList){
//                if(CollectionUtil.isNotEmpty(realTablePermission.getTableFieldInfos())){
//                    List<RealTableFieldInfoPermission> fieldInfoPermissions=realTablePermission.getTableFieldInfos();
//                    saveFieldIds.addAll(fieldInfoPermissions.stream().map(i->i.getDataId()).collect(Collectors.toList()));
//                }
//            }
//            if (!dataIds.containsAll(saveTableIds) || !fieldIds.containsAll(saveFieldIds)) {
//                return R.error("存在无权限数据，请刷新重试");
//            }
//        }
        realTableInfoService.savePermission(dataList);
        return R.ok();
    }

    /**
     * 保存数据仓库表的权限信息
     */
    @ApiOperation("保存数据仓库表的权限信息（租户）")
    @PostMapping("/saveTenantPermission")
    @SysLog(value="实时数仓授权根据租户",type= OperationTypeEnum.APPROVECODE,
            modulesType = OperationModulesEnum.SYSMODECODE,platformModuleType = ModuleEnum.AUTH_MANAGER_MODULE_CODE)
    public R saveTenantPermission( @RequestBody  List<TenantRealTablePermission> dataList) {
        realTableInfoService.saveTenantPermission(dataList);
        return R.ok();
    }


    @ApiOperation(value = "实时表列表")
    @PostMapping("/list")   //  /data-development/bdp/realTime/table/list
    public R listPage(@RequestBody DataSearchReq param) {
        PageUtils page = realTableInfoService.queryPage(param);
        return R.okWithData(page);
    }

    @ApiOperation(value = "实时用户列表")
//    @SysLog(value = "实时用户列表", type = OperationTypeEnum.SELECTCODE, modulesType = OperationModulesEnum.TABLEMODELCODE)
    @GetMapping("/user/list")   //  /data-development/bdp/realTime/table/user/list
    public R userList( ) {
        DataSearchReq param=new DataSearchReq();
        param.setLimit(9999);
        param.setPage(1);
        List<JSONObject> userList= Lists.newArrayList();
        PageUtils page = realTableInfoService.queryPage(param);
        Map<Long,String> userMap= Maps.newHashMap();
        for (Object o : page.getList()) {
            if (o != null) {
                BdpRealTableInfo table= (BdpRealTableInfo) o;
                userMap.put(table.getCreateUser(),table.getCreatePerName());
            }
        }
        for (Long userId : userMap.keySet()) {
            JSONObject user=new JSONObject();
            user.put("userId",userId);
            user.put("userName",userMap.get(userId));
            userList.add(user);
        }
        return R.okWithData(userList);
    }
    @ApiOperation(value = "实时表详情")
    @GetMapping("/detail/{id}")   //  /data-development/bdp/realTime/table/detail/1
    public R queryDetail(@PathVariable("id") Integer id) {
        BdpRealTableInfoVO data = realTableInfoService.queryDetail(id);
        return R.okWithData(data);
    }


    @Resource
    private RestHighLevelClient client;

    @Value("${elasticsearch.fieldIndexName}")
    private String fieldIndexName;

    @ApiOperation(value = "删除实时表")
    @SysLog(value="删除实时表",type= OperationTypeEnum.DELETECODE,
            modulesType = OperationModulesEnum.SYSMODECODE,platformModuleType = ModuleEnum.DATA_MAP_MODULE_CODE)
    @PostMapping("/delete")   //  /data-development/bdp/realTime/table/delete
    public R delete(@RequestBody BdpRealTableInfoVO param) {
        if (param.getId() == null) {
            return R.error("id不能为空");
        }
        Date now = new Date();
        BdpRealTableInfo table = realTableInfoService.getById(param.getId());
        if (table == null) {
            return R.error("表不存在!");
        }
        Integer num = realTableInfoService.checkProjectRelation(param.getId());
        if (num > 0) {
            return R.error("已经被项目关联不可删除!");
        }
        Long userId = ShiroUtils.getUserId();
        if (!userId.equals(table.getCreateUser())
                && !dataPermissionService.checkDataPermission(param.getId(),
                PermissionTypeEnum.REAL_TABLE.getType(), PermissionTypeEnum.DELETE.getType())
                && RealTableType.INNER.getCode().equals(table.getTableType())
        ) {
            return R.error("无删除权限");
        }
        realTableInfoService.removeById(param.getId());
        //删除es中对应的数据表&字段
        BulkRequest request = new BulkRequest();
        if (param.getId()!=null) {
            request.add(new DeleteRequest().index(fieldIndexName).id(param.getId().toString()));
            BulkResponse delete = null;
            try {
                delete = this.client.bulk(request, RequestOptions.DEFAULT);
            } catch (IOException e) {
                e.printStackTrace();
            }
            log.info("-------"+delete.status());
        }
        TableRegisterEntity registerEntity = new TableRegisterEntity();
        Date curr = new Date();
        registerEntity.setTableInfoId(param.getId());
        registerEntity.setBatchNum(curr.getTime());
        long cost = curr.getTime() - now.getTime();
        registerEntity.setCost(cost);
        registerEntity.setRunTime(curr);
        registerEntity.setRunUser(userId);
        tableRegisterService.save(registerEntity);

        return R.ok();
    }
    @ApiOperation(value = "数据预览")
    //@SysLog(value = "数据预览", type = OperationTypeEnum.SELECTCODE, modulesType = OperationModulesEnum.TABLEMODELCODE)
    @PostMapping("/view/data")   //  /data-development/bdp/realTime/table/view/data
    public R viewData(@RequestBody BdpRealTableInfoVO param) {
        if (param.getId() != null) {
            BdpRealTableInfo table = realTableInfoService.getById(param.getId());
            if (RealTableType.INNER.getCode().equals(table.getTableType())) {
                List<Object> data = LocalKafkaUtil.viewData(kafkaUrl,
                        table.getTopicName(), "json", 10);
                return R.okWithData(data);
            } else {
                ConfConnect kafka = confConnectService.selectOne(table.getResourceId());
                if (kafka != null) {
                    List<Object> data = LocalKafkaUtil.viewData(kafka.getKafkaUrl(),
                            table.getTopicName(), "json", 10);
                    return R.okWithData(data);
                } else {
                    return R.okWithData(null);
                }
            }
        }
        if (param.getResourceId() == null) {
            return R.error("资源ID不能为空");
        }
        ConfConnect kafka = confConnectService.selectOne(param.getResourceId());
        if (kafka == null) {
            return R.error("kafka信息不存在");
        }
        if (StringUtils.isEmpty(kafka.getTopic())) {
            return R.error("Topic不能为空");
        }
        List<Object> data = LocalKafkaUtil.viewData(kafka.getKafkaUrl(), kafka.getTopic(), "json", 10);
        return R.okWithData(data);
    }

    @Value("${spring.kafka.bootstrap-servers:}")
    private String kafkaUrl;
    @ApiOperation(value = "数据topic列表")
    @SysLog(value = "数据topic列表", type = OperationTypeEnum.SELECTCODE, modulesType = OperationModulesEnum.TABLEMODELCODE)
    @PostMapping("/topic/list")   //  /data-development/bdp/realTime/table/topic/list
    public R topicData(@RequestBody BdpRealTableInfoVO param) {
        if (param.getId() == null || param.getId() == 0) {
            Set data = LocalKafkaUtil.topicList(kafkaUrl);
            return R.okWithData(data);
        }
        ConfConnect kafka = confConnectService.selectOne(param.getId());
        if (kafka == null) {
            return R.error("kafka信息不存在");
        }
        Set data = LocalKafkaUtil.topicList(kafka.getKafkaUrl());
        return R.okWithData(data);
    }
}
